/*
 * The MIT License
 *
 * Copyright (c) 2011 Takahiro Hashimoto
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 */
package com.xiyuan.smartutils.client.ws.core.impl;

import com.xiyuan.smartutils.Threads;
import com.xiyuan.smartutils.client.http.HttpHeader;
import com.xiyuan.smartutils.client.ws.ErrorCode;
import com.xiyuan.smartutils.client.ws.WebSocketHandler;
import com.xiyuan.smartutils.client.ws.WebSocketPacket;
import com.xiyuan.smartutils.client.ws.client.WebSocketClient;
import com.xiyuan.smartutils.client.ws.core.BufferManager;
import com.xiyuan.smartutils.client.ws.core.WebSocketPacketResolver;
import com.xiyuan.smartutils.client.ws.core.WebsocketPacketHeader;
import com.xiyuan.smartutils.client.ws.core.handler.StreamHandler;
import com.xiyuan.smartutils.client.ws.core.handler.impl.PacketDumpStreamHandler;
import com.xiyuan.smartutils.client.ws.core.handshake.HttpHandshake;
import com.xiyuan.smartutils.client.ws.core.handshake.ProxyHandshake;
import com.xiyuan.smartutils.client.ws.core.handshake.SSLHandshake;
import com.xiyuan.smartutils.client.ws.core.proxy.Proxy;
import com.xiyuan.smartutils.client.exception.WebSocketException;
import com.xiyuan.smartutils.client.ws.util.StringUtil;

import javax.net.ssl.SSLEngineResult;
import javax.net.ssl.SSLEngineResult.Status;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;

import static com.xiyuan.smartutils.client.ws.ErrorCode.*;
import static java.nio.channels.SelectionKey.OP_READ;
import static java.nio.channels.SelectionKey.OP_WRITE;

/**
 * The WebSocket base client.
 *
 * @author xiyuan
 */
@SuppressWarnings("FieldMayBeFinal")
public abstract class WebSocketInstance implements WebSocketClient {
    private static final String WSS_S = "wss";
    private static final String WS_S = "ws";
    private static final String EMPTY = "";
    /**
     * The log.
     */
    private static final Logger log = Logger.getLogger(WebSocketInstance.class.getName());
    
    /**
     * <pre>
     * CONNECTED -> HANDSHAKE, CLOSED
     * HANDSHAKE -> WAIT, CLOSED
     * WAIT -> WAIT, CLOSED
     * CLOSED -> CONNECTED, CLOSED
     * </pre>
     * <p>
     * .
     *
     * @author xiyuan
     */
    public static enum State {
        
        /**
         * The CONNECTED.
         */
        CONNECTED,
        /**
         * The HANDSHAKE.
         */
        HANDSHAKE,
        /**
         * The WAIT.
         */
        WAIT,
        /**
         * The CLOSING.
         */
        CLOSING,
        /**
         * The CLOSED.
         */
        CLOSED;
        
        /**
         * The state map.
         */
        private static final EnumMap<State, EnumSet<State>> STATE_MAP = new EnumMap<State, EnumSet<State>>(State.class);
        
        static {// 初始化 状态 允许变更条件状态
            
            STATE_MAP.put(CONNECTED, EnumSet.of(State.HANDSHAKE, State.CLOSED));
            STATE_MAP.put(HANDSHAKE, EnumSet.of(State.WAIT, State.CLOSED));
            STATE_MAP.put(WAIT, EnumSet.of(State.WAIT, State.CLOSING, State.CLOSED));
            STATE_MAP.put(CLOSING, EnumSet.of(State.CLOSED));
            STATE_MAP.put(CLOSED, EnumSet.of(State.CONNECTED, State.CLOSED));
        }
        
        /**
         * Can transition to.
         *
         * @param state the state
         * @return true, if successful
         */
        boolean canTransitionTo(State state) {
            EnumSet<State> set = STATE_MAP.get(this);
            if (set == null) {return false;}
            return set.contains(state);
        }
        
        /**
         * Checks if is connected.
         *
         * @return true, if is connected
         */
        boolean isConnected() {
            switch (this) {
                case CONNECTED:
                case HANDSHAKE:
                case WAIT:
                    return true;
                default:
                    break;
            }
            return false;
        }
    }
    
    /**
     * The location.
     */
    protected URI location;
    
    /**
     * the URL to which to connect.
     */
    protected String path;
    
    /**
     * The use ssl.
     */
    protected boolean useSsl = false;
    
    /**
     * The ssl handshake.
     */
    protected SSLHandshake sslHandshake;
    
    /**
     * endpoint.
     */
    protected InetSocketAddress endpointAddress;
    
    /**
     * proxy.
     */
    protected Proxy proxy;
    
    /**
     * connection timeout. Unit as millisecond.
     */
    protected int connectionTimeout = 60 * 1000;
    
    /**
     * connection read timeout. Unit as millisecond.
     */
    protected int connectionReadTimeout = 60 * 1000;
    /**
     * blocking mode.
     */
    private boolean blockingMode = false;
    
    /**
     * The packet dump mode.
     */
    private int packetDumpMode;
    
    /**
     * quit flag.
     */
    volatile protected boolean quit;
    
    /**
     * subprotocol name array.
     */
    protected String[] protocols;
    
    /**
     * The server protocols.
     */
    protected String[] serverProtocols;
    
    /**
     * The buffer size.
     */
    protected int bufferSize;
    
    /**
     * The upstream buffer.
     */
    protected ByteBuffer upstreamBuffer;
    
    /**
     * The downstream buffer.
     */
    protected ByteBuffer downstreamBuffer;
    
    /**
     * The origin.
     */
    protected String origin;
    
    /**
     * The upstream queue.
     */
    protected BlockingQueue<ByteBuffer> upstreamQueue;
    
    /**
     * The pipeline
     */
    protected WebSocketPipeline pipeline;
    
    /**
     * websocket handler.
     */
    protected WebSocketHandler<WebSocketClient> handler;
    
    /**
     * The frame parser.
     */
    private WebSocketPacketResolver packetParser;
    
    /**
     * The socket.
     */
    protected SocketChannel socket;
    
    /**
     * The selector.
     */
    protected Selector selector;
    
    /**
     * The handshake.
     */
    private HttpHandshake handshake;
    
    /**
     * The response header map.
     */
    protected HttpHeader responseHeader;
    
    /**
     * The request header map.
     */
    protected HttpHeader requestHeader;
    
    /**
     * The http response status.
     */
    protected int responseStatus;
    
    /**
     * The websocket state.
     */
    protected volatile State state = State.CLOSED;
    
    /**
     * The close latch.
     */
    protected CountDownLatch closeLatch;
    
    /**
     * The handshake latch.
     */
    protected CountDownLatch handshakeLatch;
    
    /**
     * worker.
     */
    protected ExecutorService executorService;
    
    private AtomicInteger executorThreadNumber = new AtomicInteger(0);
    
    private String executorThreadName;
    
    private int autoPingDelay = 5;
    
    /**
     * Instantiates a new web socket base.
     *
     * @param url       the url
     * @param handler   the handler
     * @param protocols the protocols
     * @throws WebSocketException the web socket exception
     */
    public WebSocketInstance(String url, WebSocketHandler<WebSocketClient> handler,
                             String... protocols) throws WebSocketException {
        this.init(url, handler);
        this.protocols = protocols;
        // Override origin
        this.origin = this.location.getHost() + (this.location.getPort() > 0 ? ":" + this.location.getPort() : "");
        
    }
    
    /**
     * Instantiates a new web socket base.
     *
     * @param url       the url
     * @param handler   the handler
     * @param protocols the protocols
     * @throws WebSocketException the web socket exception
     */
    public WebSocketInstance(String url, String origin, WebSocketHandler<WebSocketClient> handler,
                             String... protocols) throws WebSocketException {
        init(url, handler);
        this.origin = origin;
        this.protocols = protocols;
    }
    
    /**
     * Instantiates a new web socket base.
     *
     * @param url       the url
     * @param origin    the origin
     * @param proxy     the proxy
     * @param handler   the handler
     * @param protocols the protocols
     * @throws WebSocketException the web socket exception
     */
    public WebSocketInstance(String url, String origin, Proxy proxy, WebSocketHandler<WebSocketClient> handler,
                             String... protocols) throws WebSocketException {
        this.protocols = protocols;
        this.proxy = proxy;
        this.init(url, handler);
        this.origin = origin;
        
    }
    
    /**
     * Instantiates a new web socket base.
     *
     * @param url       the url
     * @param proxy     the proxy
     * @param handler   the handler
     * @param protocols the protocols
     * @throws WebSocketException the web socket exception
     */
    public WebSocketInstance(String url, Proxy proxy, WebSocketHandler<WebSocketClient> handler,
                             String... protocols) throws WebSocketException {
        this.protocols = protocols;
        this.proxy = proxy;
        this.init(url, handler);
        this.origin = this.location.getHost() + ":" + this.location.getPort();
    }
    
    /**
     * Inits the.
     *
     * @param url the url
     * @throws WebSocketException the web socket exception
     */
    protected void init(String url, WebSocketHandler<WebSocketClient> handler) throws WebSocketException {
        
        // init properties
        initializeProperties();
        // parse url
        parseUrl(url);
        
        // init pipe
        initializePipeline();
        
        // init handler
        this.handler = handler;
    }
    
    /**
     * Initialize properties.
     *
     * @throws WebSocketException the web socket exception
     */
    protected void initializeProperties() throws WebSocketException {
        int upstreamQueueSize = Integer.getInteger("websocket.upstreamQueueSize", 500);
        this.bufferSize = Integer.getInteger("websocket.bufferSize", 0x7FFF);
        this.packetDumpMode = Integer.getInteger("websocket.packatdump", 0);
        this.upstreamQueue = new LinkedBlockingQueue<ByteBuffer>(upstreamQueueSize);
        
        this.downstreamBuffer = ByteBuffer.allocate(this.bufferSize);
        this.upstreamBuffer = ByteBuffer.allocate(this.bufferSize);
        this.requestHeader = new HttpHeader();
    }
    
    /**
     * Initialize pipeline.
     *
     * @throws WebSocketException the web socket exception
     */
    protected void initializePipeline() throws WebSocketException {
        
        if (this.useSsl) {this.sslHandshake = new SSLHandshake(this.endpointAddress, this);}
        
        // setup pipeline
        this.pipeline = new WebSocketPipeline(new StreamHandler() {
            private PacketDumpStreamHandler dumpStreamHandler = new PacketDumpStreamHandler();
            
            @Override
            public void write(WebSocketClient ws, ByteBuffer netBuffer) throws WebSocketException {
                // 2、’ dump upstream
                int pos = netBuffer.position(), limit = netBuffer.limit();
                this.dumpStreamHandler.write(ws, netBuffer);
                // 还原数据位置（因为不知道dumpStream 会不会改变）
                netBuffer.position(pos).limit(limit);
                
                // 3、 send remote
                if (!upstreamQueue.offer(netBuffer)) {throw new WebSocketException(E3030);}
                
                selector.wakeup();
            }
            
            @Override
            public void read(WebSocketClient ws, ByteBuffer localBuffer) throws WebSocketException {
                
                // 2、 dump upstream
                int newPos = localBuffer.position(), limit = localBuffer.limit();
                this.dumpStreamHandler.read(ws, localBuffer);
                // 还原数据位置（因为不知道dumpStream 会不会改变）
                localBuffer.position(newPos).limit(limit);
                
                // 3、 parse WebSocket parcket
                WebSocketPacket packet = getPacketParser().resolve(localBuffer);
                if (packet == null) {return;}
                
                WebsocketPacketHeader header = packet.getHeader();
                if (header.isPing()) {handler.onPing(ws);}
                else if (header.isPong()) {handler.onPong(ws);}
                else if (header.isClose()) {
                    try {
                        handler.onCloseFrame(ws, packet);
                    }
                    finally {
                        stateTo(State.CLOSED);
                    }
                }
                else {
                    handler.onMessage(ws, packet);
                }
            }
            
            @Override
            public void writeHandshake(WebSocketClient ws, ByteBuffer netBuffer) throws WebSocketException {
                // 2、’ dump upstream
                int pos = netBuffer.position(), limit = netBuffer.limit();
                this.dumpStreamHandler.writeHandshake(ws, netBuffer);
                // 还原数据位置（因为不知道dumpStream 会不会改变）
                netBuffer.position(pos).limit(limit);
                
                // 3、 send remote
                if (!upstreamQueue.offer(netBuffer)) {throw new WebSocketException(E3030);}
                selector.wakeup();
            }
            
            @Override
            public void readHandshake(WebSocketClient ws, ByteBuffer localBuffer) throws
                                                                                  WebSocketException {
                
                // 2、’ dump upstream
                int pos = localBuffer.position(), limit = localBuffer.limit();
                this.dumpStreamHandler.readHandshake(ws, localBuffer);
                // 还原数据位置（因为不知道dumpStream 会不会改变）
                localBuffer.position(pos).limit(limit);
                
                // 3 parse http header.
                HttpHandshake hs = getHandshake();
                hs.handshakeResponse(localBuffer);
                // 3 通知处理器打开
                if (hs.isDone()) {handler.onOpen(ws);}
            }
        });
        
    }
    
    /**
     * Parses the url.
     *
     * @param urlStr the url str
     * @throws WebSocketException the web socket exception
     */
    private void parseUrl(String urlStr) throws WebSocketException {
        try {
            URI uri = new URI(urlStr);
            if (!(WS_S.equals(uri.getScheme()) || WSS_S.equals(uri.getScheme()))) {
                throw new WebSocketException(E3007, uri.toString());
            }
            
            if (WSS_S.equals(uri.getScheme())) {
                useSsl = true;
            }
            String query = uri.getQuery();
            path = (EMPTY.equals(uri.getPath()) ? "/" : uri.getPath()) + (query != null ? "?" + query : "");
            int port = uri.getPort();
            if (port < 0) {
                if (WS_S.equals(uri.getScheme())) {
                    port = 80;
                }
                else if (WSS_S.equals(uri.getScheme())) {
                    port = 443;
                    useSsl = true;
                }
                else {
                    throw new WebSocketException(E3008, uri.toString());
                }
            }
            // new SocketAdd
            endpointAddress = new InetSocketAddress(uri.getHost(), port);
            location = uri;
        }
        catch (URISyntaxException e) {
            throw new WebSocketException(E3009, e);
        }
    }
    
    /*
     * (non-Javadoc)
     *
     * @see jp.a840.websocket.WebSocket#getLocation()
     */
    @Override
    public URI getLocation() {
        return location;
    }
    
    /*
     * (non-Javadoc)
     *
     * @see jp.a840.websocket.WebSocket#send(jp.a840.websocket.frame.Frame)
     */
    @Override
    public void send(WebSocketPacket packet) throws WebSocketException {
        if (!isConnected()) {throw new WebSocketException(E3010);}
        
        pipeline.upstream(this, packet.toByteBuffer());
    }
    
    /**
     * Send.
     *
     * @param obj the obj
     * @throws WebSocketException the web socket exception
     */
    public void send(Object obj) throws WebSocketException {
        send(createPacket(obj));
    }
    
    /*
     * (non-Javadoc)
     *
     * @see jp.a840.websocket.WebSocket#isConnected()
     */
    @Override
    public boolean isConnected() {
        return state.isConnected();
    }
    
    /*
     * (non-Javadoc)
     *
     * @see jp.a840.websocket.WebSocket#close()
     */
    @Override
    public void close() {
        try {
            this.quit = true;
            if (state != State.WAIT) {return;}
            onClose();
            stateTo(State.CLOSED);
            selector.wakeup();
            
            if (!Thread.currentThread().getName().equals(this.executorThreadName)) {
                try {
                    closeLatch.await(30, TimeUnit.SECONDS);
                }
                catch (InterruptedException e) {
                
                }
            }
            
            try {
                if (executorService != null) {
                    executorService.shutdown();
                    executorService.awaitTermination(30, TimeUnit.SECONDS);
                }
            }
            catch (InterruptedException e) {
                ;
            }
            finally {
                synchronized (this) {
                    executorService = null;
                }
            }
        }
        catch (WebSocketException e) {
            // handler.onError(this, e);
        }
    }
    
    @Override
    public void connect() throws WebSocketException {
        try {
            // 1. check connection status
            if (isConnected()) {throw new WebSocketException(E3039);}
            
            if (!this.state.canTransitionTo(State.CONNECTED)) {throw new WebSocketException(E3040, state.name());}
            
            // 2. initialize connection
            handshakeLatch = new CountDownLatch(1);
            closeLatch = new CountDownLatch(1);
            
            // 3 create socket.
            selector = Selector.open();
            socket = SocketChannel.open();
            socket.configureBlocking(false);
            socket.register(selector, OP_READ);
            
            // 4、connect server.
            long start = System.currentTimeMillis();
            InetSocketAddress remoteAddress = this.endpointAddress;
            ProxyHandshake proxyHandshake = null;
            if (proxy != null) {proxyHandshake = this.proxy.getProxyHandshake(this);}
            
            if (proxyHandshake != null) {remoteAddress = proxyHandshake.getProxyAddress();}
            
            // 4.1 start connect to remote address
            if (socket.connect(remoteAddress)) {throw new WebSocketException(E3041);}
            
            // 5. check timeout.
            while (!socket.finishConnect()) {
                if ((System.currentTimeMillis() - start) > connectionTimeout) {throw new WebSocketException(E3042);}
                Threads.sleepIgnoreException(10);
            }
            // 6. connect done try handshakes
            stateTo(State.CONNECTED);
            if (proxyHandshake != null) {proxyHandshake.doHandshake(socket);}
            
            // 7 SSL/TLS doHandshake
            if (useSsl) {sslHandshake.doHandshake(socket);}
            
            // 8. send Http handshake request
            // String str = "GET / HTTP/1.1\r\n"
            // + "Host: 127.0.0.1\r\n"
            // + "Upgrade: websocket\r\n"
            // + "Connection: Upgrade\r\n"
            // + "Sec-WebSocket-Key: ciu1zNziQ/q22vK+Au1VEg==\r\n"
            // + "Origin: 127.0.0.1:80\r\n"
            // + "Sec-WebSocket-Version: 13\r\n"
            // + "\r\n"
            // + "";
            pipeline.upstreamHandshake(this, getHandshake().createHandshakeRequest());
            stateTo(State.HANDSHAKE);
            
            final WebSocketInstance ws = this;
            Runnable worker = new Runnable() {
                private BufferManager bufferManager = new BufferManager();
                
                public void run() {
                    
                    long time = System.currentTimeMillis();
                    long sec = 0;
                    while (!quit) {
                        try {
                            selector.select(1000);
                            
                            Set<SelectionKey> keys = selector.selectedKeys();
                            Iterator<SelectionKey> it = keys.iterator();
                            while (it.hasNext()) {
                                SelectionKey key = it.next();
                                it.remove();
                                
                                SocketChannel channel = (SocketChannel) key.channel();
                                if (key.isValid() && key.isWritable() && upstreamQueue.peek() != null) {
                                    ByteBuffer netBuffer = wrap(upstreamQueue.poll(), E3819);
                                    channel.write(netBuffer);
                                }
                                else if (key.isValid() && key.isReadable()) {
                                    // read
                                    downstreamBuffer.clear();
                                    if (channel.read(downstreamBuffer) < 0) {throw new WebSocketException(E3020);}
                                    
                                    downstreamBuffer.flip();
                                    ByteBuffer localBuffer = unwrap(downstreamBuffer, E3820);
                                    switch (state) {
                                        case HANDSHAKE:
                                            // CONNECTED -> HANDSHAKE
                                            pipeline.downstreamHandshake(ws, localBuffer);
                                            if (getHandshake().isDone()) {
                                                handshakeLatch.countDown();
                                                pipeline.downstream(ws, localBuffer);
                                                stateTo(State.WAIT);
                                            }
                                            break;
                                        case WAIT:
                                        case CLOSING:
                                            pipeline.downstream(ws, localBuffer);
                                            break;
                                    }
                                }
                            }
                            
                            int op = ws.state != State.CLOSED ? OP_READ | OP_WRITE : OP_WRITE;
                            if (!upstreamQueue.isEmpty()) {socket.register(ws.selector, op);}
                            if (state == State.CLOSED) {
                                quit = true;
                            }
                            
                            long time1 = System.currentTimeMillis();
                            if (!quit && time1 - time >= 1000) {
                                sec += (time1 - time) / 1000;
                                time = time1;
                                if (sec >= autoPingDelay && state == State.WAIT) {
                                    try {
                                        ping();
                                        sec = 0;
                                    }
                                    catch (Exception e) {}
                                }
                                
                            }
                        }
                        catch (WebSocketException we) {
                            handler.onError(ws, we);
                        }
                        catch (IOException ioe) {
                            quit = true;
                            handler.onError(ws, new WebSocketException(E3020, ioe));
                        }
                        catch (Throwable e) {
                            handler.onError(ws, new WebSocketException(E3043, e));
                        }
                    }
                    
                    try {
                        socket.close();
                    }
                    catch (IOException ex) {
                        log.log(Level.INFO, "close socket error", ex);
                    }
                    try {
                        selector.close();
                    }
                    catch (IOException ex) {
                        log.log(Level.INFO, "close selector error", ex);
                    }
                    
                    try {
                        if (state == State.WAIT) {
                            handler.onClose(ws);
                        }
                    }
                    catch (Throwable e) {}
                    handshakeLatch.countDown();
                    closeLatch.countDown();
                    handshake = null;
                    packetParser = null;
                }
                
                /**
                 * wraped buffer if use ssl.
                 * @param localBuffer the local data
                 * @param code        the error code if ssl engine is closed
                 * @return the net buffer of the socket.
                 * @throws WebSocketException
                 */
                private ByteBuffer wrap(ByteBuffer localBuffer, ErrorCode code) throws WebSocketException {
                    ByteBuffer netBuffer = !useSsl ? localBuffer : ByteBuffer.allocate(getBufferSize());
                    
                    // ssl wrap
                    while (useSsl && localBuffer.hasRemaining()) {
                        if (netBuffer.position() != 0) {// 扩容
                            ByteBuffer temp = ByteBuffer.allocate(netBuffer.flip().remaining() + getBufferSize());
                            temp.put(netBuffer);
                            netBuffer = temp;
                        }
                        
                        SSLEngineResult res = sslHandshake.wrap(localBuffer, netBuffer);
                        if (res.getStatus() == Status.CLOSED) {throw new WebSocketException(code);}
                    }
                    if (useSsl) {netBuffer.flip();}
                    
                    return netBuffer;
                }
                
                /**
                 *
                 * @param netBuffer net data
                 * @param code      the error code if ssl engine is closed
                 * @return the local buffer of app. or null if fragmented
                 *         netBuffer
                 * @throws WebSocketException exception
                 */
                public ByteBuffer unwrap(ByteBuffer netBuffer, ErrorCode code) throws WebSocketException {
                    netBuffer = bufferManager.merge(netBuffer);
                    int pos = netBuffer.position();
                    
                    ByteBuffer localBuffer = netBuffer;
                    if (useSsl) {localBuffer = ByteBuffer.allocate(netBuffer.remaining() + getBufferSize());}
                    
                    boolean unwraped = true;
                    // ssl umwap
                    while (useSsl && netBuffer.hasRemaining()) {
                        SSLEngineResult res = sslHandshake.unwrap(netBuffer, localBuffer);
                        if (res.getStatus() == Status.CLOSED) {throw new WebSocketException(code);}
                        
                        if (res.getStatus() == Status.BUFFER_UNDERFLOW) {// 当前网络数据包不足以解析
                            unwraped = false;
                            break;
                        }
                        
                        if (res.getStatus() == Status.BUFFER_OVERFLOW) {// 目标缓冲区不足以容纳，解析过后的数据包(目标数据包需要 扩容)
                            ByteBuffer temp = ByteBuffer.allocate(localBuffer.flip().remaining() + getBufferSize());
                            temp.put(localBuffer);
                            localBuffer = temp;
                        }
                    }
                    
                    if (!unwraped) {// 没有完全解包 则需要保存下一次重新解包
                        netBuffer.position(pos);
                        bufferManager.storeFragmentBuffer(netBuffer);
                        return null;
                    }
                    
                    // 切换读模式
                    if (useSsl) {
                        localBuffer.flip();
                        bufferManager.reset();// 重置缓存
                    }
                    
                    return localBuffer;
                }
                
            };
            
            this.quit = false;
            if (blockingMode) {
                worker.run();
            }
            else {
                this.executorThreadName = "WebSocket-Client-" + executorThreadNumber.incrementAndGet();
                Thread t = new Thread(worker, executorThreadName);
                t.setDaemon(true);
                t.start();
                handshakeLatch.await();
                // 等待线程 结束
                // this.closeLatch.await();
                // this.executorService.shutdown();
            }
            
        }
        catch (Throwable e) {
            try {
                socket.close();
            }
            catch (IOException ex) {
                log.log(Level.INFO, "close socket error", ex);
            }
            try {
                selector.close();
            }
            catch (IOException ex) {
                log.log(Level.INFO, "close selector error", ex);
            }
            throw e instanceof WebSocketException ? (WebSocketException) e : new WebSocketException(E3044, e);
        }
    }
    
    
    /**
     * Transition to.
     *
     * @param to the to
     * @return the state
     */
    protected State stateTo(State to) {
        if (state.canTransitionTo(to)) {
            State old = state;
            state = to;
            return old;
        }
        else {
            throw new IllegalStateException("Couldn't transtion from " + state + " to " + to);
        }
    }
    
    /**
     * State.
     *
     * @return the state
     */
    protected State state() {
        return state;
    }
    
    /**
     * Creates the socket.
     *
     * @return the socket channel
     * @throws IOException Signals that an I/O exception has occurred.
     */
    protected SocketChannel createSocket() throws IOException {
        SocketChannel socket = SocketChannel.open();
        socket.configureBlocking(false);
        return socket;
    }
    
    public void setAutoPingDelay(int delay) {
        if (delay < 1) return;
        this.autoPingDelay = delay;
    }
    
    /**
     * 获取自动保持心跳时长
     */
    public int getAutoPingDelay() {
        return this.autoPingDelay;
    }
    
    /**
     * Close web socket.
     *
     * @throws WebSocketException the web socket exception
     */
    protected abstract void onClose() throws WebSocketException;
    
    /**
     * Gets the web socket version.
     *
     * @return the web socket version
     */
    protected abstract int getWebSocketVersion();
    
    /**
     * New handshake instance.
     *
     * @return the handshake
     */
    protected abstract HttpHandshake newHandshakeInstance();
    
    /**
     * New WebSocketPacketParser instance.
     *
     * @return the handshake
     */
    protected abstract WebSocketPacketResolver newPacketParserInstance();
    
    /**
     * Gets the handshake.
     *
     * @return the handshake
     */
    protected synchronized HttpHandshake getHandshake() {
        if (handshake == null) {handshake = newHandshakeInstance();}
        
        return handshake;
    }
    
    /**
     * Gets the frame parser.
     *
     * @return the frame parser
     */
    protected synchronized WebSocketPacketResolver getPacketParser() {
        if (packetParser == null) {packetParser = newPacketParserInstance();}
        return packetParser;
    }
    
    
    /**
     * (non-Javadoc)
     */
    public boolean isBlockingMode() {
        return blockingMode;
    }
    
    @Override
    public void setBlockingMode(boolean blockingMode) {
        this.blockingMode = blockingMode;
    }
    
    /**
     * Gets the server protocols.
     *
     * @return the server protocols
     */
    public String[] getServerProtocols() {
        return serverProtocols;
    }
    
    /**
     * Sets the server protocols.
     *
     * @param serverProtocols the new server protocols
     */
    public void setServerProtocols(String[] serverProtocols) {
        this.serverProtocols = serverProtocols;
    }
    
    /**
     * Gets the path.
     *
     * @return the path
     */
    public String getPath() {
        return path;
    }
    
    
    /**
     * Gets the protocols.
     *
     * @return the protocols
     */
    public String[] getProtocols() {
        return protocols;
    }
    
    /**
     * Gets the origin.
     *
     * @return the origin
     */
    public String getOrigin() {
        return origin;
    }
    
    @Override
    public String getHeader(String name) {
        return requestHeader == null ? null : requestHeader.getHeader(name);
    }
    
    @Override
    public List<String> getHeaders(String name) {
        return requestHeader == null ? null : requestHeader.getHeaders(name);
    }
    
    public InetSocketAddress getEndpoint() {
        return endpointAddress;
    }
    
    /**
     * Gets the response header.
     *
     * @return the response header
     */
    public HttpHeader getResponseHeader() {
        return responseHeader;
    }
    
    /**
     * Gets the request header.
     *
     * @return the request header
     */
    public HttpHeader getRequestHeader() {
        return requestHeader;
    }
    
    /**
     * Gets the response status.
     *
     * @return the response status
     */
    public int getResponseStatus() {
        return responseStatus;
    }
    
    public int getConnectionTimeout() {
        return connectionTimeout;
    }
    
    /**
     * Set The Connection Timeout. Unit as second.
     *
     * @param connectionTimeout Unit as second.
     */
    public void setConnectionTimeout(int connectionTimeout) {
        if (connectionTimeout < 3) {return;}
        this.connectionTimeout = connectionTimeout * 1000;
    }
    
    /**
     * Gets the connection read timeout. Unit as second.
     *
     * @return the connection read timeout
     */
    public int getConnectionReadTimeout() {
        return connectionReadTimeout / 1000;
    }
    
    /**
     * Sets the connection read timeout.
     *
     * @param connectionReadTimeout the new connection read timeout
     */
    public void setConnectionReadTimeout(int connectionReadTimeout) {
        if (connectionReadTimeout < 1) {return;}
        
        this.connectionReadTimeout = connectionReadTimeout * 1000;
    }
    
    /**
     * Sets the origin.
     *
     * @param origin the new origin
     */
    public void setOrigin(String origin) {
        this.origin = origin;
    }
    
    public int getBufferSize() {
        return bufferSize;
    }
    
    /**
     * Gets the packet dump mode.
     *
     * @return the packet dump mode
     */
    public int getPacketDumpMode() {
        return this.packetDumpMode;
    }
    
    /**
     * Sets the packet dump mode.
     *
     * @param packetDumpMode the new packet dump mode
     */
    public void setPacketDumpMode(int packetDumpMode) {
        this.packetDumpMode = packetDumpMode;
    }
    
    /*****************************************/
    // 提供子类调用
    /*****************************************/
    
    /**
     * Join.
     *
     * @param delim       the delim
     * @param collections the collections
     * @return the string
     */
    protected static String join(String delim, Collection<String> collections) {
        return StringUtil.join(delim, collections);
    }
    
    /**
     * Join.
     *
     * @param delim   the delim
     * @param strings the strings
     * @return the string
     */
    protected static String join(String delim, String... strings) {
        return StringUtil.join(delim, strings);
    }
    
    /**
     * Join.
     *
     * @param delim   the delim
     * @param start   the start
     * @param end     the end
     * @param strings the strings
     * @return the string
     */
    protected static String join(String delim, int start, int end, String... strings) {
        return StringUtil.join(delim, start, end, strings);
    }
}
